package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime220815.bean.PlayVideoTimesBean;
import com.atguigu.edu.realtime220815.bean.PlayVideoViewingDucBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author: rym
 * @Description: TODO
 * @DateTime: 2023/2/17 15:58
 * 累积播放时长
 **/
public class DwsPlayVideoViewingDurationWindow {
    public static void main(String[] args) throws Exception {
        /*
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        /*create table appVideo_log(
                common map<String,String>,
                appVideo map<String,String>,
                ts bigint,
                order_time as to_timestamp(from_unixtime(ts/1000)),
                WATERMARK FOR order_time AS order_time - INTERVAL '2' SECOND
)
        tableEnv.executeSql("create table appVideo_log(\n" +
                "common map<String,String>,\n" +
                "appVideo map<String,String>,\n" +
                "ts bigint,\n" +
                "order_time as to_timestamp(from_unixtime(ts/1000)),\n" +
                "WATERMARK FOR order_time AS order_time - INTERVAL '2' SECOND\n" +
                ")"+ KafkaUtils.getKafkaSqlWith("dwd_traffic_appVideo_log","DwsPlayVideoViewingDurationWindow_group"));

        select
        appVideo['video_id'],
                appVideo['play_sec'],
                order_time
        from appVideo_log
        where appVideo['play_sec'] is not null

        Table filTable = tableEnv.sqlQuery("select \n" +
                "appVideo['video_id'] video_id,\n" +
                "appVideo['play_sec'] play_sec,\n" +
                "order_time\n" +
                "from appVideo_log\n" +
                "where appVideo['play_sec'] is not null");

        tableEnv.createTemporaryView("filTable",filTable);
        //tableEnv.executeSql("select * from filTable").print();
        /*
        * +----+--------------------------------+--------------------------------+-------------------------+
| op |                       video_id |                       play_sec |              order_time |
+----+--------------------------------+--------------------------------+-------------------------+
| +I |                            710 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                            710 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                            710 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                            710 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                           5410 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                           5410 |                             30 | 2023-02-17 16:36:21.000 |
| +I |                            710 |                             30 | 2023-02-17 16:36:21.000 |*/
        /*
        Table groupByTable = tableEnv.sqlQuery("select \n" +
                "date_format(TUMBLE_START(order_time, INTERVAL '10' second) , 'yyyy-MM-dd HH:mm:ss') stt,\n" +
                "date_format(TUMBLE_END(order_time, INTERVAL '10' second), 'yyyy-MM-dd HH:mm:ss') edt,\n" +
                "video_id,\n" +
                "sum(play_sec),\n" +
                "UNIX_TIMESTAMP()*1000 ts\n" +
                "from filTable\n" +
                "group by video_id,TUMBLE(order_time, INTERVAL '10' second)");
        tableEnv.createTemporaryView("groupByTable",groupByTable);
        tableEnv.executeSql("select * from groupByTable");*/

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_traffic_appVideo_log";
        String groupID = "DwsPlayVideoViewingDurationWindow_groupId";
        FlinkKafkaConsumer<String> flinkKafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupID);
        DataStreamSource<String> kafkaDS = env.addSource(flinkKafkaConsumer);

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);


        SingleOutputStreamOperator<PlayVideoViewingDucBean> beanDs = jsonObjDS.map(
                new MapFunction<JSONObject, PlayVideoViewingDucBean>() {
                    @Override
                    public PlayVideoViewingDucBean map(JSONObject value) throws Exception {
                        Double play_sec = value.getJSONObject("appVideo").getDouble("play_sec");
                        String video_id = value.getJSONObject("appVideo").getString("video_id");
                        Long ts = value.getLong("ts");
                        PlayVideoViewingDucBean playVideoViewingDucBean = PlayVideoViewingDucBean.builder()
                                .play_sec_total(play_sec)
                                .video_id(video_id)
                                .ts(ts)
                                .build();
                        return playVideoViewingDucBean;
                    }
                }
        );

        KeyedStream<PlayVideoViewingDucBean, String> keyDS = beanDs.keyBy(value -> value.getVideo_id());

        WindowedStream<PlayVideoViewingDucBean, String, TimeWindow> windowDS = keyDS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<PlayVideoViewingDucBean> reduceDS = windowDS.reduce(
                new ReduceFunction<PlayVideoViewingDucBean>() {
                    @Override
                    public PlayVideoViewingDucBean reduce(PlayVideoViewingDucBean value1, PlayVideoViewingDucBean value2) throws Exception {
                        value1.setPlay_sec_total(value1.getPlay_sec_total() + value2.getPlay_sec_total());
                        return value1;
                    }
                },
                new WindowFunction<PlayVideoViewingDucBean, PlayVideoViewingDucBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<PlayVideoViewingDucBean> input, Collector<PlayVideoViewingDucBean> out) throws Exception {
                            long start = window.getStart();
                            long end = window.getEnd();
                            long now = System.currentTimeMillis();
                        for (PlayVideoViewingDucBean playVideoViewingDucBean : input) {
                            playVideoViewingDucBean.setStt(DateFormatUtil.toYmdHms(start));
                            playVideoViewingDucBean.setEdt(DateFormatUtil.toYmdHms(end));
                            playVideoViewingDucBean.setTs(now);
                            out.collect(playVideoViewingDucBean);
                        }
                    }
                }
        );

        reduceDS.print();

        SingleOutputStreamOperator<PlayVideoViewingDucBean> JoinDS1 = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<PlayVideoViewingDucBean>("dim_video_info") {

                    @Override
                    public void join(JSONObject dimInfoJsonObj, PlayVideoViewingDucBean obj) {
                        String chapter_id = dimInfoJsonObj.getString("chapter_id".toUpperCase());
                        obj.setChapter_id(chapter_id);
                    }

                    @Override
                    public String getKey(PlayVideoViewingDucBean obj) {
                        return obj.getVideo_id();
                    }
                }, 60, TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<PlayVideoViewingDucBean> JoinDS2 = AsyncDataStream.unorderedWait(
                JoinDS1,
                new DimAsyncFunction<PlayVideoViewingDucBean>("dim_chapter_info") {

                    @Override
                    public void join(JSONObject dimInfoJsonObj, PlayVideoViewingDucBean obj) {
                        String chapter_name = dimInfoJsonObj.getString("chapter_name".toUpperCase());
                        obj.setChapter_name(chapter_name);
                    }

                    @Override
                    public String getKey(PlayVideoViewingDucBean obj) {
                        return obj.getChapter_id();
                    }
                }, 60, TimeUnit.SECONDS
        );

        //JoinDS2.print();
        JoinDS2.addSink(ClickHouseUtils.getClickHouseSinkFunction(
                "insert into dws_play_video_viewing_duration_window values(?,?,?,?,?,?,?)"
        ));

        env.execute();

    }






}
